package com.flc.archive.mq.consumer;

import com.flc.archive.config.AppConfig;
import com.flc.archive.mq.listener.ReportListener;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author haizi
 * @date 2018/1/23
 */
public class ReportConsumer implements Consumer {

    private static final Logger logger = LoggerFactory.getLogger(ReportConsumer.class);

    @Override
    public void start() throws MQClientException {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(ReportConsumer.class.getSimpleName());

        AppConfig.RocketMqCfg rocketMqCfg = AppConfig.getInstance().getRocketMqCfg();

        consumer.setNamesrvAddr(rocketMqCfg.getNameServerAddr());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.subscribe(rocketMqCfg.getReportTopic(), "*");

        consumer.registerMessageListener(new ReportListener());

        consumer.start();

        logger.info(ReportConsumer.class.getName() + " Started!!!");
    }
}
